package com.yeziji.dataCenter.business.dataSource.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSONObject;
import com.mybatisflex.core.datasource.DataSourceKey;
import com.mybatisflex.core.row.DbChain;
import com.mybatisflex.core.row.Row;
import com.yeziji.common.business.dataMessage.base.DataMessage;
import com.yeziji.common.business.dataMessage.service.DataMessageRecordService;
import com.yeziji.constant.DataMessageTopicEnum;
import com.yeziji.constant.DataOperateTypeEnum;
import com.yeziji.constant.DataSyncStatusEnum;
import com.yeziji.dataCenter.business.dataSource.service.DataOperateService;
import com.yeziji.dataCenter.common.msg.DataCenterErrorMsg;
import com.yeziji.devops.business.dataSource.entity.DevopsDataSourceEntity;
import com.yeziji.devops.business.dataSource.service.DevopsDataSourceService;
import com.yeziji.devops.common.msg.DevopsErrorMsg;
import com.yeziji.exception.ApiException;
import com.yeziji.utils.RabbitMqUtils;
import com.yeziji.utils.expansion.Asserts;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.Map;
import java.util.Set;

/**
 * @author hwy
 * @since 2024/10/26 20:37
 **/
@Slf4j
@Service
public class DataOperateServiceImpl implements DataOperateService {
    @Resource
    private DevopsDataSourceService devopsDataSourceService;
    @Resource
    private DataMessageRecordService dataMessageRecordService;

    @Override
    public void execute(DataMessage dataMessage) {
        final String curDataSourceKey = DataSourceKey.get();
        String objMessage = dataMessage.getMessage();
        JSONObject jsonObject = JSONObject.parseObject(objMessage);
        DevopsDataSourceEntity devopsDataSource = Asserts.notNull(devopsDataSourceService.getByName(dataMessage.getFlag()), DevopsErrorMsg.DATA_SOURCE_IS_NOT_EXISTS);
        try {
            String table = dataMessage.getTable();
            DataSourceKey.use(devopsDataSource.getName());
            DataOperateTypeEnum operateTypeEnum = DataOperateTypeEnum.getByCode(dataMessage.getOperateType());
            switch (operateTypeEnum) {
                case INSERT:
                    this.insert(DbChain.table(table), jsonObject);
                    break;
                case UPDATE:
                    this.update(DbChain.table(table), dataMessage.getId(), jsonObject);
                    break;
                case DELETE:
                    jsonObject.put("is_delete", 1);
                    this.update(DbChain.table(table), dataMessage.getId(), jsonObject);
                    break;
                default:
                    Asserts.error(DataCenterErrorMsg.NOT_SUPPORTED_DB_TYPE);
            }
            // 发送消息确认变更
            jsonObject.put("sync_status", DataSyncStatusEnum.SUCCESS.getCode());
        } catch (Exception e) {
            jsonObject.put("sync_status", DataSyncStatusEnum.FAILED.getCode());
            log.error("同步数据异常: {}", e.getMessage(), e);
            throw e;
        } finally {
            if (StrUtil.isNotBlank(curDataSourceKey)) {
                DataSourceKey.use(curDataSourceKey);
            } else {
                DataSourceKey.clear();
            }
            // 将消息确认结果发送给客户端
            dataMessage.setMessage(jsonObject.toJSONString());
            dataMessageRecordService.sendTo(
                    DataMessageTopicEnum.PLATFORM_DATA_CONFIRM_SYNC.getExchange(),
                    RabbitMqUtils.getAsRabbitPlatformKey(devopsDataSource.getCreator(), DataMessageTopicEnum.PLATFORM_DATA_CONFIRM_SYNC.getRouteKey()),
                    dataMessage,
                    null
            );
        }
    }

    private void insert(DbChain dbChain, JSONObject jsonObject) {
        Set<Map.Entry<String, Object>> rowEntries = getAsRowByJson(jsonObject).entrySet();
        for (Map.Entry<String, Object> rowEntry : rowEntries) {
            dbChain.set(rowEntry.getKey(), rowEntry.getValue());
        }
        dbChain.save();
    }

    private void update(DbChain dbChain, Long id, JSONObject jsonObject) {
        Row row = getAsRowByJson(jsonObject);
        Integer version = row.getInt("version");
        boolean update =
                dbChain.where(queryWrapper -> {
                    queryWrapper.eq("id", id)
                            // 数据中心中，更新的数据要低一个版本
                            .eq("version", version - 1);
                }).update(row);
        if (!update) {
            throw new ApiException(DataCenterErrorMsg.DATA_VERSION_CONFLICT);
        }
    }

    private static Row getAsRowByJson(JSONObject jsonObject) {
        Row row = new Row();
        row.toUnderlineKeysMap();
        for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            // 时间格式，外部传过来序列化后成为了 Long
            if (key.contains("_time")) {
                row.put(key, new Date((Long) value));
            } else {
                row.put(key, value);
            }
        }
        return row;
    }
}
